package com.ruoyi.smart.mqtt;

import com.ruoyi.smart.config.MqttProperties;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.SmartLifecycle;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;

@Slf4j
@Component
public class MqttClient implements SmartLifecycle, MqttCallbackExtended {

    @Autowired
    private MqttProperties mqttProperties;

    private MqttAsyncClient mqttAsyncClient;
    private MqttConnectOptions connectOptions;
    private boolean isRunning = false;
    private int reconnectDelay = 1;
    private final ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();

    /**
     * 消息处理器映射表: Key -> TopicFilter, Value -> MessageHandler实例
     */
    private final Map<String, MessageHandler> messageHandlerMap = new ConcurrentHashMap<>();

    @PostConstruct
    public void init() {
        try {
            connectOptions = new MqttConnectOptions();
            connectOptions.setUserName(mqttProperties.getUsername());
            connectOptions.setPassword(mqttProperties.getPassword().toCharArray());
            connectOptions.setCleanSession(true);
            connectOptions.setConnectionTimeout(mqttProperties.getConnectionTimeout());
            connectOptions.setKeepAliveInterval(mqttProperties.getKeepAliveInterval());
            connectOptions.setAutomaticReconnect(mqttProperties.isAutomaticReconnect());

            mqttAsyncClient = new MqttAsyncClient(
                    mqttProperties.getServerUri(),
                    mqttProperties.getClientId(),
                    new MemoryPersistence()
            );

            mqttAsyncClient.setCallback(this);
        } catch (MqttException e) {
            log.error("MQTT客户端初始化失败", e);
        }
    }

    /**
     * 注册消息处理器
     * @param handler 消息处理器实例
     */
    public void registerMessageHandler(MessageHandler handler) {
        messageHandlerMap.put(handler.getTopicFilter(), handler);
        log.info("注册MQTT消息处理器: {}", handler.getTopicFilter());

        // 如果已经连接，立即订阅该主题
        if (mqttAsyncClient != null && mqttAsyncClient.isConnected()) {
            try {
                mqttAsyncClient.subscribe(handler.getTopicFilter(), 1);
                log.info("已订阅主题: {}", handler.getTopicFilter());
            } catch (MqttException e) {
                log.error("订阅主题失败: {}", handler.getTopicFilter(), e);
            }
        }
    }

    @Override
    public void start() {
        if (!isRunning) {
            connect();
            isRunning = true;
            log.info("MQTT客户端已启动");
        }
    }

    @Override
    public void stop() {
        if (isRunning) {
            disconnect();
            isRunning = false;
            log.info("MQTT客户端已停止");
        }
    }

    @Override
    public boolean isRunning() {
        return isRunning;
    }

    private void connect() {
        try {
            log.info("尝试连接到MQTT服务器: {}", mqttProperties.getServerUri());
            mqttAsyncClient.connect(connectOptions, null, new IMqttActionListener() {
                @Override
                public void onSuccess(IMqttToken asyncActionToken) {
                    log.info("MQTT连接成功");
                    reconnectDelay = 1; // 重置重连延迟
                    subscribeAllHandlers(); // 连接成功后订阅所有处理器主题
                }

                @Override
                public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
                    log.error("MQTT连接失败", exception);
                    scheduleReconnect();
                }
            });
        } catch (MqttException e) {
            log.error("MQTT连接异常", e);
            scheduleReconnect();
        }
    }

    /**
     * 订阅所有已注册处理器的主题
     */
    private void subscribeAllHandlers() {
        if (messageHandlerMap.isEmpty()) {
            log.warn("没有注册任何消息处理器，将不会订阅任何主题");
            return;
        }

        try {
            for (String topicFilter : messageHandlerMap.keySet()) {
                mqttAsyncClient.subscribe(topicFilter, 1);
                log.debug("已订阅主题: {}", topicFilter);
            }
            log.info("已成功订阅 {} 个主题", messageHandlerMap.size());
        } catch (MqttException e) {
            log.error("订阅主题失败", e);
        }
    }

    private void disconnect() {
        try {
            scheduler.shutdown();
            if (mqttAsyncClient != null && mqttAsyncClient.isConnected()) {
                mqttAsyncClient.disconnect();
            }
        } catch (MqttException e) {
            log.error("MQTT断开连接失败", e);
        }
    }

    private void scheduleReconnect() {
        log.info("计划在 {} 秒后重新连接", reconnectDelay);
        scheduler.schedule(this::connect, reconnectDelay, TimeUnit.SECONDS);
        reconnectDelay = Math.min(reconnectDelay * 2, mqttProperties.getMaxReconnectDelay());
    }

    /**
     * 发布消息到指定主题
     * @param topic 主题
     * @param message 消息内容
     * @param qos QoS等级
     * @param retained 是否保留消息
     */
    public void publish(String topic, String message, int qos, boolean retained) {
        if (mqttAsyncClient != null && mqttAsyncClient.isConnected()) {
            try {
                MqttMessage mqttMessage = new MqttMessage(message.getBytes());
                mqttMessage.setQos(qos);
                mqttMessage.setRetained(retained);
                mqttAsyncClient.publish(topic, mqttMessage);
                log.debug("消息发布成功: topic={}", topic);
            } catch (MqttException e) {
                log.error("发布消息失败: topic={}", topic, e);
            }
        } else {
            log.warn("MQTT客户端未连接，无法发布消息: topic={}", topic);
        }
    }

    // ============ MqttCallbackExtended 接口实现 ============

    @Override
    public void connectComplete(boolean reconnect, String serverURI) {
        log.info("MQTT连接{}成功", reconnect ? "重连" : "");
        reconnectDelay = 1;
        if (reconnect) {
            subscribeAllHandlers(); // 重连后重新订阅所有主题
        }
    }

    @Override
    public void connectionLost(Throwable cause) {
        log.warn("MQTT连接丢失", cause);
        scheduleReconnect();
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        // 【统一处理开始】- 所有消息都会经过这里
        long startTime = System.currentTimeMillis();
        log.debug("收到消息: 主题={}, QoS={}, 内容长度={}",
                topic, message.getQos(), message.getPayload().length);

        try {
            // 1. 查找匹配的消息处理器
            MessageHandler handler = findHandlerForTopic(topic);

            if (handler != null) {
                // 2. 找到处理器，将消息分发出去
                handler.handleMessage(topic, message);
            } else {
                // 3. 没有找到处理器，记录警告
                log.warn("未找到匹配的消息处理器 for topic: {}", topic);
            }

        } catch (Exception e) {
            // 【统一异常处理】- 捕获所有处理器抛出的异常
            log.error("处理MQTT消息时发生异常: topic={}", topic, e);
        } finally {
            // 性能监控
            long costTime = System.currentTimeMillis() - startTime;
            if (costTime > 1000) {
                log.warn("MQTT消息处理耗时较长: {}ms, topic: {}", costTime, topic);
            }
        }
    }

    /**
     * 根据实际主题查找匹配的处理器
     */
    private MessageHandler findHandlerForTopic(String actualTopic) {
        for (Map.Entry<String, MessageHandler> entry : messageHandlerMap.entrySet()) {
            String topicFilter = entry.getKey();
            if (matchesTopic(topicFilter, actualTopic)) {
                return entry.getValue();
            }
        }
        return null;
    }

    /**
     * 主题匹配逻辑（支持+单级通配符）
     */
    private boolean matchesTopic(String filter, String actualTopic) {
        log.debug("filter:{} actualTopic:{}", filter, actualTopic);

        if (filter.equals(actualTopic)) {
            return true;
        }

        return MqttTopicMatcher.matches(filter, actualTopic);
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        log.debug("消息发布完成: messageId={}", token.getMessageId());
    }

    @PreDestroy
    public void destroy() {
        stop();
    }
}
